package com.example.grpc.client.controller;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.protobuf.Message;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.examples.hellostreaming.DubboStreamingGreeterGrpc;
import io.grpc.examples.hellostreaming.HelloReply;
import io.grpc.examples.hellostreaming.HelloRequest;
import io.grpc.stub.StreamObserver;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Slf4j
@RestController
public class GRpcDemo2Controller {

    @DubboReference(interfaceClass = DubboStreamingGreeterGrpc.IStreamingGreeter.class,protocol = "grpc",url = "grpc://172.16.36.76:29080",timeout=5000)
    private DubboStreamingGreeterGrpc.IStreamingGreeter streamingGreeter;

    @GetMapping("sayHelloTen")
    public String sayHelloTen(){
        HelloRequest weiwei = HelloRequest.newBuilder().setName("weiwei").build();
        Iterator<HelloReply> replys = null;
        List<String> list = new ArrayList<>();
        try {
            replys = streamingGreeter.sayHelloTen(weiwei);
            for (int i = 1; replys.hasNext(); i++) {
                HelloReply reply = replys.next();
                log.info("Result # {} : {}",i, reply);
                list.add(reply.getMessage());
            }
        } catch (StatusRuntimeException e) {
            log.error("RPC failed: {}", e.getStatus());
        }
        return list.toString();
    }

    @SneakyThrows
    @GetMapping("/sayHelloStream")
    public String sayHelloStream(){
        List<String> result = Lists.newArrayList();
        CountDownLatch countDownLatch = new CountDownLatch(1);
        StreamObserver<HelloRequest> requestObserver = streamingGreeter.sayHelloStreaming(new StreamObserver<HelloReply>() {
            @Override
            public void onNext(HelloReply helloReply) {
                log.info("获取到消息：{}",helloReply.getMessage());
                result.add(helloReply.getMessage());
            }

            @Override
            public void onError(Throwable throwable) {
                log.warn("sayHelloStream Failed : {}", Status.fromThrowable(throwable));
                countDownLatch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("Finish sayHelloStream");
                countDownLatch.countDown();
            }
        });

        for (int i=1;i<11;i++){
            String name = "weiwei -- " + i;
            log.info("发送 {}--{}","weiwei",i);
            HelloRequest req = HelloRequest.newBuilder().setName(name).build();
            requestObserver.onNext(req);
        }
        requestObserver.onCompleted();
        if (!countDownLatch.await(1, TimeUnit.MINUTES)) {
            log.warn("routeChat can not finish within 1 minutes");
        }
        return result.toString();
    }

    @SneakyThrows
    @GetMapping("/sayHelloStreamGetOne")
    public String sayHelloStreamGetOne(){
        List<String> result = Lists.newArrayList();
        CountDownLatch countDownLatch = new CountDownLatch(1);
        StreamObserver<HelloRequest> requestObserver = streamingGreeter.sayHelloStreamingGetOneReply(new StreamObserver<HelloReply>() {
            @Override
            public void onNext(HelloReply helloReply) {
                log.info("获取到消息：{}",helloReply.getMessage());
                result.add(helloReply.getMessage());
            }

            @Override
            public void onError(Throwable throwable) {
                log.warn("sayHelloStream Failed : {}", Status.fromThrowable(throwable));
                countDownLatch.countDown();
            }

            @Override
            public void onCompleted() {
                log.info("Finish sayHelloStream");
                countDownLatch.countDown();
            }
        });

        for (int i=1;i<11;i++){
            String name = "weiwei -- " + i;
            log.info("发送 {}--{}","weiwei",i);
            HelloRequest req = HelloRequest.newBuilder().setName(name).build();
            requestObserver.onNext(req);
        }
        requestObserver.onCompleted();
        if (!countDownLatch.await(1, TimeUnit.MINUTES)) {
            log.warn("routeChat can not finish within 1 minutes");
        }
        return result.toString();
    }

    @VisibleForTesting
    interface TestHelper {
        /**
         * Used for verify/inspect message received from server.
         */
        void onMessage(Message message);

        /**
         * Used for verify/inspect error received from server.
         */
        void onRpcError(Throwable exception);
    }

}
